package com.chief.hadoop.mr.join.map;

import com.chief.hadoop.mr.join.OrderBean;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.viewfs.ViewFileSystem;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;

public class OrderMapper extends Mapper<LongWritable, Text, OrderBean, OrderBean> {

    OrderBean outKey = new OrderBean();

    Map<String, String> goods = new HashMap<>();

    @Override
    protected void setup(Context context) throws IOException {
        Path path = new Path(context.getCacheFiles()[0]);
        FileSystem fileSystem = ViewFileSystem.newInstance(context.getConfiguration());
        FSDataInputStream fsDataInputStream = fileSystem.open(path);
        BufferedReader reader = new BufferedReader(new InputStreamReader(fsDataInputStream, "UTF-8"));
        String line;
        while (StringUtils.isNotBlank(line = reader.readLine())) {
            String[] s = line.split(" ");
            goods.put(s[0], s[1]);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] split = value.toString().split(" ");
        outKey.setGoodsName(goods.get(split[1]));
        outKey.setGoodsId(split[1]);
        outKey.setOrderId(split[0]);
        outKey.setCnt(Integer.valueOf(split[2]));
        context.write(outKey, outKey);
    }
}
